package com.jml.demo.simple;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 客户端： 消费消息
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //构建消费者实例，蚕丝是消费者组，相同的组名的多个消费者实例，搭建消费集群
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //指定namerserver地址，不是broker
        consumer.setNamesrvAddr("172.26.3.145:9876");
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //指定topic进行消息的订阅，全部 *，第二个参数是根据tag
        //一个topic下面有多个tag，可以指定消费某个tag，也可以指消费全部tag
        consumer.subscribe("TopicTest", "*");
//        consumer.setMaxReconsumeTimes(18);
//        consumer.getConsumeThreadMin();
        //注册消息监听器
        //MessageListenerConcurrently适用多线程兵法操作
        //MessageListenerOrderly适用顺序接受，一个队列只能被一个线程接受
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //第一个参数，消息内容的集合，一次接受多个消息
            // 第二个参数 消费的上下文。
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //接收的RocketMQ发送的消息
                for (MessageExt msg:msgs){
                    //消费消息，是多线程实现  ！
                    System.out.printf("%s Receive New Messages: %s %n",
                            Thread.currentThread().getName(), new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //消费启动
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
